from pyspark import SparkContext, SparkConf

from pyspark.sql.session import SparkSession

conf = SparkConf().setMaster("local").setAppName("homepage")
sc = SparkContext(conf=conf)
spark = SparkSession.builder.appName('homepage').getOrCreate()
import datetime
from elasticsearch import Elasticsearch

es = Elasticsearch(hosts='106.13.117.37', port='9200')
ll = []
body = {
    "query": {
        "match_all": {}
    }
}
data = es.search(index="movieinfo", body=body, size=20000).get('hits').get('hits')
for i in data:
    ll.append(i['_source'])

rdd = sc.parallelize(ll)
df = spark.createDataFrame(rdd)

# 计算票房同比昨日增长率
today = '"%s"' % (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
yesterday = '"%s"' % (datetime.date.today() - datetime.timedelta(days=2)).strftime("%Y-%m-%d")

t_df = df.filter('date==%s' % today)
t_sell = t_df.groupby().sum().collect()[0].asDict()['sum(boxOffice)']

y_df = df.filter('date==%s' % yesterday)
y_sell = y_df.groupby().sum().collect()[0].asDict()['sum(boxOffice)']

sell_rate = (t_sell - y_sell) / y_sell

# 计算上座率同比昨日增长率
t_sit = t_df.groupby().avg().collect()[0].asDict()['avg(attendance)']
y_sit = y_df.groupby().avg().collect()[0].asDict()['avg(attendance)']
sit_rate = (t_sit - y_sit) / y_sit

# 计算本月票房增长率
from pyspark.sql.functions import udf, col
from pyspark.sql.types import StringType

my_func = udf(lambda a: a[0:7], StringType())
df = df.withColumn('month', my_func('date'))
m_df = df.groupby("month").sum().sort("month").collect()
currnet = (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m")
month_rate = 0
all_month = [i.asDict() for i in m_df]
for i in range(len(all_month)):
    if all_month[i]['month'] == currnet:
        month_rate = (all_month[i]['sum(boxOffice)'] - all_month[i - 1]['sum(boxOffice)']) / all_month[i - 1][
            'sum(boxOffice)']
        break

# 计算本月票房数量：
m_sell = '"%s"' % (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m")
month_sell = df.filter('month == %s' % m_sell)
month_sell = month_sell.groupby().sum().collect()[0].asDict()['sum(boxOffice)']

# 计算今日榜首电影信息
day_info = t_df.sort("boxOffice").collect()[-1].asDict()
day_name = day_info['movie_name']
day_sell = day_info['boxOffice']
day_image = day_info['movie_img']

# 计算本月榜首电影信息

month_info = \
    [i.asDict() for i in
     df.filter('month == %s' % m_sell).groupby("movie_name").sum().sort("sum(boxOffice)").collect()][-1]
month_name = month_info['movie_name']
month_sell = month_info['sum(boxOffice)']
month_image = df.filter('movie_name == "%s"' % month_name).collect()[-1].asDict()['movie_img']

# 计算该月每日票房 ，计算该月每日上座率
every_month_info = df.filter('month == %s' % m_sell).groupby("date").avg().sort("date").collect()
every_sell = [i.asDict()['avg(boxOffice)'] for i in every_month_info]
every_sit = [i.asDict()['avg(attendance)'] for i in every_month_info]

# 计算年榜，返回前四名
s_year = '"%s-01-01"' % (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y")
e_year = '"%s-12-31"' % (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y")
year = df.filter('date >= %s and date<=%s' % (s_year, e_year)).groupby("movie_name").sum().sort(
    "sum(boxOffice)").collect()[:6]
year_info = [i.asDict() for i in year]
name = [i['movie_name'] for i in year_info]
year_rank = []
for i in name:
    image = df.filter('movie_name == "%s"' % i).collect()[0].asDict()['movie_img']
    temp_body = {
        "query": {
            "match": {
                "movie_name": i
            }
        }
    }
    rank = es.search(index='moviecomment', body=temp_body).get('hits').get('hits')[0]['_source']['movie_score'][
        'movie_score']
    year_rank.append({'name': i, 'image': image, 'rank': rank})

# 写入数据库
body = {
    'date': datetime.date.today() - datetime.timedelta(days=1),
    'sell_rate': sell_rate,
    'sit_rate': sit_rate,
    'month_rate': month_rate,
    'day': {'day_name': day_name, 'day_sell': day_sell, 'day_image': day_image},
    'month': {'month_name': month_name, 'month_sell': month_sell, 'month_image': month_image},
    'every_sell': every_sell,
    'every_sit': every_sit,
    'year': year_rank
}
es.index(index='homepage', body=body)

